// @Author EthanScriptOn
// @Desc
package dispatcher

import (
	"fmt"
	"gitee.com/fatzeng/srf_switch_basic_components/queue"
	"gitee.com/fatzeng/srf_switch_basic_components/safe"
	"gitee.com/fatzeng/srf_switch_basic_components/tool"
	"math"
)

//MonitoringBridge The monitor role in the Data Orchestrator
type MonitoringBridge struct {
	monitor    *queue.QueueLoadMonitor
	queue      queue.Queue
	bridgeName DispatcherEnum
}
type MonitoringBridgePool map[*MonitoringBridge]struct{}

// DataDispatcher The data coordinator is responsible for forwarding data between different queues
type DataDispatcher struct {
	config       *DispatcherConfig                       // config Data orchestrator configuration
	bridge       map[DispatcherEnum]MonitoringBridgePool // bridge key-Bridge name val-Bridge monitoring
	crossLinkMap map[string]uint64                       // crossLinkMap Record the relationship between the bridges
	share        bool                                    // share Whether or not to broadcast to consumers
	mu           *tool.ReentryLock                       // mu Reentrant lock
	increment    chan struct{}                           // increment Start the annunciator
	safe         *safe.SafeGo                            // Abnormal recovery
}

func GenerateDataDispatcher(config *DispatcherConfig, share bool) (*DataDispatcher, error) {
	dispatcher := &DataDispatcher{
		config:       config,
		bridge:       make(map[DispatcherEnum]MonitoringBridgePool),
		crossLinkMap: make(map[string]uint64),
		share:        share,
		mu:           tool.GenerateReentryLock(false), // Adopt the principle of unfairness
		increment:    make(chan struct{}),
		safe:         safe.GenerateSafeGo(),
	}
	if err := dispatcher.safe.SafeGo(dispatcher.incrementProcess); err != nil {
		return nil, err
	}
	return dispatcher, nil
}

// Register Sign up for a bridge
func (d *DataDispatcher) Register(name DispatcherEnum, queueInstance queue.Queue) {
	_ = tool.LockReentryWithSync(d.mu, func() error {
		queues, ok := d.bridge[name]
		if !ok {
			queues = make(MonitoringBridgePool)
		}
		monitoringBridge := new(MonitoringBridge)
		monitoringBridge.monitor = queue.GenerateQueueLoadMonitor(queueInstance)
		monitoringBridge.queue = queueInstance
		monitoringBridge.bridgeName = name
		queues[monitoringBridge] = struct{}{}
		d.bridge[name] = queues
		return nil
	})
	d.increment <- struct{}{}
}

// incrementProcess Listen for the registration operation and start processing
func (d *DataDispatcher) incrementProcess() {
	for {
		select {
		case _, ok := <-d.increment:
			if ok {
				_ = d.Start()
			}
		}
	}
}

// Start Turn on processing
func (d *DataDispatcher) Start() error {
	return d.safe.SafeGo(func() {
		_ = tool.LockReentryWithSync(d.mu, func() error {
			for _, componentName := range d.config.componentsName {
				// As a producer, you need to check whether the number of producers
				// that have been opened is consistent with the target quantity
				maybeProducer := componentName
				consumerName, err := d.config.GetConsumer(maybeProducer)
				if err == nil {
					producerMonitors, hasProducerMonitors := d.bridge[maybeProducer]
					consumerMonitors, hasConsumerMonitors := d.bridge[consumerName]
					if hasProducerMonitors &&
						hasConsumerMonitors &&
						len(producerMonitors) > 0 &&
						len(consumerMonitors) > 0 &&
						d.metering(producerMonitors, consumerMonitors, true) {
						d.crossLink(consumerMonitors, producerMonitors)
					}
				}
				// As a consumer, check whether the number of consumers
				// that have been opened is consistent with the target number
				maybeConsumer := componentName
				producerName, err := d.config.GetProducer(maybeConsumer)
				if err == nil {
					producerMonitors, hasProducerMonitors := d.bridge[producerName]
					consumerMonitors, hasConsumerMonitors := d.bridge[maybeConsumer]
					if hasProducerMonitors &&
						hasConsumerMonitors &&
						len(producerMonitors) > 0 &&
						len(consumerMonitors) > 0 &&
						d.metering(producerMonitors, consumerMonitors, false) {
						d.crossLink(consumerMonitors, producerMonitors)
					}
				}
			}
			return nil
		})
	})
}

// metering Measure whether the quantity of producers and consumers is eligible
func (d *DataDispatcher) metering(producerMonitors, consumerMonitors MonitoringBridgePool, contrast bool) bool {
	var isMetering bool
	_ = tool.LockReentryWithSync(d.mu, func() error {
		for producerBridgeName := range producerMonitors {
			for consumerBridgeName := range consumerMonitors {
				// I look for it from the connection map of the bridge
				contract := d.crossLinkMap[fmt.Sprintf("%v-%v", producerBridgeName, consumerBridgeName)]
				if contrast {
					isMetering = uint64(len(producerMonitors)) > contract
				} else {
					isMetering = uint64(len(consumerMonitors)) > contract
				}
				return nil
			}
		}
		return nil
	})
	return isMetering
}

// crossLink Build bridges to connect with each other
func (d *DataDispatcher) crossLink(producerMonitors MonitoringBridgePool, consumerMonitors MonitoringBridgePool) {
	_ = tool.LockReentryWithSync(d.mu, func() error {
		for producerQueue := range producerMonitors {
			// Start the renewal before processing the data and complete some of the pre-required operations
			survivalConsumerMonitors := d.findSurvivalConsumerMonitors(consumerMonitors)
			if len(survivalConsumerMonitors) <= 0 {
				continue
			}
			// Metering +1
			for _, consumerQueueTemp := range survivalConsumerMonitors {
				d.signContract(fmt.Sprintf("%v-%v", producerQueue.bridgeName, consumerQueueTemp.bridgeName))
			}
			// Enable GO to bridge data stream forwarding
			if err := d.safe.SafeGo(func() {
				func(producerQueueTemp *MonitoringBridge) {
					// Data forwarding
					d.taskProcessing(producerQueueTemp, survivalConsumerMonitors)
					if producerQueueTemp.queue.IsClose() {
						delete(d.bridge[producerQueueTemp.bridgeName], producerQueueTemp)
					}
					for _, consumerQueueTemp := range survivalConsumerMonitors {
						d.terminateContract(fmt.Sprintf("%v-%v", producerQueueTemp.bridgeName, consumerQueueTemp.bridgeName))
					}
				}(producerQueue)
			}); err != nil {
				return err
			}
		}
		return nil
	})
}

// taskProcessing Task processing, task distribution
func (d *DataDispatcher) taskProcessing(producerMonitor *MonitoringBridge, consumerMonitors []*MonitoringBridge) {
	for pullData, err := producerMonitor.queue.Pull(); err == nil; {
		if pullData == nil {
			continue
		}
		// Producers open a one-to-many processing model for consumers
		if d.share {
			for _, consumerQueue := range consumerMonitors {
				consumerQueue.queue.Push(pullData)
			}
		} else {
			// Open the one-to-one processing mode to find the right consumer
			var minFactor = math.MaxFloat64
			var queueInstance queue.Queue
			for _, consumerQueue := range consumerMonitors {
				currentLoadFactor := consumerQueue.monitor.GetLoadPercentage()
				if currentLoadFactor < minFactor {
					minFactor = currentLoadFactor
					queueInstance = consumerQueue.queue
				}
			}
			if queueInstance != nil {
				// Message forwarding
				queueInstance.Push(pullData)
			}
		}
	}
}

// signContract Metering +1
func (d *DataDispatcher) signContract(crossLinkKey string) {
	_ = tool.LockReentryWithSync(d.mu, func() error {
		contract, ok := d.crossLinkMap[crossLinkKey]
		if !ok {
			d.crossLinkMap[crossLinkKey] = 1
		} else {
			contract++
			d.crossLinkMap[crossLinkKey] = contract
		}
		return nil
	})
}

// terminateContract Meter-1
func (d *DataDispatcher) terminateContract(crossLinkKey string) {
	_ = tool.LockReentryWithSync(d.mu, func() error {
		contract, ok := d.crossLinkMap[crossLinkKey]
		if ok {
			contract--
			d.crossLinkMap[crossLinkKey] = contract
		}
		return nil
	})
}

// findSurvivalConsumerMonitors Find the right consumers
func (d *DataDispatcher) findSurvivalConsumerMonitors(monitors MonitoringBridgePool) []*MonitoringBridge {
	survivalConsumerMonitors := make([]*MonitoringBridge, 0)
	_ = tool.LockReentryWithSync(d.mu, func() error {
		for consumerQueue := range monitors {
			consumerQueueTemp := consumerQueue
			if consumerQueueTemp.queue.IsClose() {
				delete(monitors, consumerQueueTemp)
				continue
			}
			survivalConsumerMonitors = append(survivalConsumerMonitors, consumerQueueTemp)
		}
		return nil
	})
	return survivalConsumerMonitors
}
